// Copyright (c) 2000 Just Objects B.V. <just@justobjects.nl>
// Distributable under LGPL license. See terms of license at gnu.org.

package nl.justobjects.pushlet.core;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Hashtable;

/**
 * Subscribes to events from a Publisher.
 *
 * @version $Id: PulletSubscriber.java,v 1.4 2003/08/15 08:37:40 justb Exp $
 * @author Just van den Broecke - Just Objects &copy;
 **/
public class PulletSubscriber extends HTTPSubscriber {
	private static Hashtable pushletPullSubscribers = new Hashtable(32);
	private String myId;
	long aliveTimeOut = 15000;
	volatile long lastAlive = System.currentTimeMillis();

	public PulletSubscriber(String id) {
		myId = id;
		pushletPullSubscribers.put(id, this);
	}

	public static PulletSubscriber get(String id) {
		return (PulletSubscriber) pushletPullSubscribers.get(id);
	}

	protected boolean hasBailedOut() {
		return myId == null;
	}

	/** Extended bailing out: also remove ourselves from collection. */
	protected void bailout() {
		if (hasBailedOut()) {
			return;
		}

		// Unsubscribe from Publisher
		super.bailout();

		// Remove ourselves from collection
		if (pushletPullSubscribers.containsKey(myId)) {
			pushletPullSubscribers.remove(myId);
			myId = null;
		}
	}

	protected ClientAdapter createClientAdapter(HttpServletRequest httpServletRequest,
												HttpServletResponse httpServletResponse) throws IOException {

		String outputFormat = getFormat();

		// Default assume a browser window on the other end.
		if (outputFormat.equals(FORMAT_JAVASCRIPT)) {
			return BrowserClientAdapter.createPulletAdapter(httpServletRequest, httpServletResponse, myId);
		}
		else if (outputFormat.equals(FORMAT_XML)) {
			// return  new JavaPushletAdapter();
			return null; // TODO support Java clients
		}
		else {
			return null;
		}
	}

	/** Called when subsequent events need to be processed. */
	public void initContinue(HttpServletRequest httpServletRequest,
								HttpServletResponse httpServletResponse) throws IOException {
		if (clientAdapter != null) {
			clientAdapter.init(httpServletRequest, httpServletResponse);
			clientAdapter.start();
		}
		else {
			warn("clientAdapter == null");
		}
	}


	/** Get events from queue and push to clients. */
	public void processEvents() throws IOException {

		Event nextEvent = null;

		// Main loop: each time a event is received from the Publisher it is written to document.
		do {
			// Get next event; blocks (wait()) until event is returned
			try {
				nextEvent = (Event) eventQueue.deQueue();
			} catch (InterruptedException ie) {
				warn("interrupted");
			}

			if (nextEvent == null) {
				warn("nothing from Queue, bailing out: " + mySubject);
				bailout();
				return;
			}

			// Let clientAdapter determine how to send event
			try {
				// p("push: "+nextEvent.getSubject());
				clientAdapter.push(nextEvent);
			}
			catch (IOException ioe) {
				// Client may have closed connection.
				// Remove ourselves from Publisher and return.
				// e("register for leave: "+mySubject);
				bailout();
				return;
			}
			catch (Throwable t) {
				warn("register for leave: " + t + " subj=" + mySubject);
				bailout();
				return;
			}

			// Indicate we are still alive and pumping
			lastAlive = System.currentTimeMillis();

		}
		while (!eventQueue.isEmpty() && (!hasBailedOut()));  // send at least one event

		// Send file postamble (triggers new GET).
		clientAdapter.stop();
	}

	/** Event from Publisher: enqueue it. */
	synchronized public void send(Event theEvent) {
		// p("send: queue event: "+theEvent.getSubject());

		// Check if we had any active continuation for at
		// least 'timeOut' millisecs. If the client has left this
		// instance there would be no way of knowing otherwise.
		// Put event in queue; may block if queue full
		long now = System.currentTimeMillis();
		if (now - lastAlive > aliveTimeOut) {
			warn("not alive for at least: " + aliveTimeOut + "ms leaving...");
			bailout();
			return;
		}

		super.send(theEvent);

		// Event in queue; see sendEvents() where it is handled
		return;
	}

	/** Exceptional print util. */
	public void warn(String s) {
		Log.warn("PulletSubscriber-" + this + ":" + s);
	}
}

/*
 * $Log: PulletSubscriber.java,v $
 * Revision 1.4  2003/08/15 08:37:40  justb
 * fix/add Copyright+LGPL file headers and footers
 *
 * Revision 1.3  2003/08/12 09:57:05  justb
 * replaced all print statements to Log.*() calls
 *
 * Revision 1.2  2003/05/18 16:15:08  justb
 * support for XML encoded Events
 *
 * Revision 1.1.1.1  2002/09/24 21:02:32  justb
 * import to sourceforge
 *
 * Revision 1.1.1.1  2002/09/20 22:48:18  justb
 * import to SF
 *
 * Revision 1.1.1.1  2002/09/20 14:19:04  justb
 * first import into SF
 *
 * Revision 1.4  2002/04/15 20:42:41  just
 * reformatting and renaming GuardedQueue to BlockingQueue
 *
 * Revision 1.3  2000/10/30 14:15:47  just
 * no message
 *
 * Revision 1.2  2000/08/21 20:48:29  just
 * added CVS log and id tags plus copyrights
 *
 *
 */
